package com.aotain.nyx.flink;

import java.util.HashSet;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;


import org.apache.log4j.Logger;

import com.aotain.apollo.IPDatabase;
import com.aotain.apollo.IPSearcher;
import com.aotain.common.CommonFunction;
import com.aotain.common.GlobalLock;
import com.aotain.nyx.abnormal.AbnormalCheckImport;
import com.aotain.nyx.util.IPFilterSingleUtil;
import com.aotain.nyx.util.IPFilterUtil;

public class SourceDataFilterMap implements MapFunction<String, Tuple2<String,String>> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 6137189732292460920L;


	private Integer Number = -1;
			
//	static HashSet<String> IPSet;
//	static HashSet<String> WhiteIPSet;
	final long timeInterval = 1000*15*60;
	String filterIpSql ;
	String whiteIpSql;
	String dbJsonNYX;
	String dbJsonSDS;
	
	/**
	 * 
	 * @param filterIpSql
	 * @param whiteIpSql
	 * @param dbJsonsds SDS
	 * @param dbJsonnyx NYX
	 */
	public SourceDataFilterMap(final String filterIpSql,
			final String whiteIpSql,
			final String dbJsonsds,
			final String dbJsonnyx) {
		this.filterIpSql = filterIpSql;
		this.whiteIpSql = whiteIpSql;
		this.dbJsonNYX = dbJsonnyx;
		this.dbJsonSDS = dbJsonsds;
		Runnable runnable = new Runnable() {  
			public void run() {  
				while (true) {  
					try {  
//						IPSet = IPFilterUtil.getFilterIpSet(filterIpSql,dbJson);
//						WhiteIPSet = IPFilterUtil.getWhiteIpSet(whiteIpSql, dbJson);
						IPFilterSingleUtil.instance = null;
//						System.out.println("WihteIPSet update over ! Set'size is " + IPSet.size());  
						Thread.sleep(timeInterval);  
					} catch (InterruptedException e) {  
						e.printStackTrace();  
					}  
				}  
			}  
		};  
		Thread thread = new Thread(runnable);  
		thread.start();  
		
		Number = GlobalLock.getInscance().GetNumber();
	}

	@Override
	public Tuple2<String,String> map(String value) throws Exception {
		// TODO Auto-generated method stub

		String refer = "";             String sourceAreaCountry = null;    String domainname = "";                    
		String dip = "";               String sourceAreaCity = null;       String url = "";                           
		String sip = "";               String sourceAreaName = null;       String type = "";                          
		String userAgent = "";         String destAreaCountry = null;      String accesstime = "";         
		String cookie = "";            String destAreaCity = null;         String idcno = "";              
		String dipL = "";              String destAreaName = null;         String post = "0";	          
		String sipL = "";              String gisstr = null;               String dsip = "";               
		String dport = "";             String gisstrBase64 = null;         String areaName_d = null;       
		String sport = "";             String Sourcelon = null;            long upStreamOctets = 0;        
		int status ;                   long upStreamPacket = 0;            String areaName_s = null;           
		String Sourcelat = null;       String sourceAreaId = null;         String destAreaId = null;          
		String Destlon = null;         String sourceAreaProvinceId = null; String destAreaProvinceId = null;  
		String Destlat = null;         String sourceAreaCityId = null;     String destAreaCityId = null;      
		String delaytime = null;	   long sipLong;					    long dipLong;
		long dnStreamPacket = 0;       long dnStreamOctets = 0;				String appType;
		long accesstimeL = 0l;
		String appId;
		Tuple2<String,String> ret = new Tuple2<String,String>();

		String[] splits = value.split("\\|",-1);
		//		System.out.println("***value" + value);
		if(splits.length>22) {
			String userAccount = splits[1];
			String ipRex = "([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}";   
			Pattern pattern = Pattern.compile(ipRex);   
			Matcher matcher = pattern.matcher(userAccount);  
			dipL = splits[9];
			dport = splits[10];
			sipL = splits[12];
			sport = splits[13];
			type = splits[7]; 
			accesstime= splits[14];
			try {
				upStreamOctets = Long.valueOf(splits[18]);
				upStreamPacket = Long.valueOf(splits[20]);
				dnStreamOctets = Long.valueOf(splits[17]);
				dnStreamPacket = Long.valueOf(splits[19]);
			} catch (Exception ex) {
				Logger.getLogger(SourceDataFilterMap.class).error("DATA Format ERROR:" + value,ex);
			}
			
			appType = splits[5];
			appId = splits[6];
			idcno = "446605";
			if("1".equals(splits[22]) && "6".equals(splits[7]) && splits.length >43) {//http
				//if(splits.length ==43) {//http
				domainname = splits[32];
				url = domainname + "/" + splits[28];
				status = Integer.valueOf(splits[23]);
				userAgent = splits[26];
				refer = splits[37];
				cookie = splits[42];
				if(splits.length>43) {
					post  = splits[43];
				}
			}else{//非Http
				domainname = "";
				url = "";
				status = Integer.valueOf(splits[23]);
			}
			
			sipLong = Long.parseLong(sipL);
			dipLong = Long.parseLong(dipL);
			dip = CommonFunction.int2ip(dipLong);
			sip = CommonFunction.int2ip(sipLong);
			
				
			dsip = dip+"#"+sip;
			delaytime = 0 + "";
			long currenttime = System.currentTimeMillis();
			if(accesstime != null && !"".equals(accesstime)) {
				delaytime = currenttime/1000 - Long.parseLong(accesstime.trim()) + "";
				accesstimeL = Long.parseLong(accesstime.trim());  //水位线使用
			}
			
			if(!IPFilterSingleUtil.getInstance(filterIpSql, whiteIpSql, dbJsonNYX).WhiteIpSet.contains(dsip)) {

				if(!"".equals(dipL) && !"".equals(dport) && !"".equals(sipL) && !"".equals(sport) ) {

					IPSearcher ipsearcher = IPSearcher.getInstance(dbJsonSDS,Number);
					IPDatabase SourceArea = ipsearcher.SearchIPDB(sip);
					IPDatabase DestArea = ipsearcher.SearchIPDB(dip);
					HashSet<String> IPSet = IPFilterSingleUtil.getInstance(filterIpSql, whiteIpSql, dbJsonNYX).IPSet;
					
					
					if(SourceArea != null) {
						sourceAreaCountry = SourceArea.getCountryName();
						sourceAreaCity = SourceArea.getCityName();
						//						sourceAreaProvince = SourceArea.getProviceName();
						if(sourceAreaCity != null) {
							sourceAreaName  = sourceAreaCity;
							sourceAreaId = SourceArea.getCityID()+"";
						} else {
							sourceAreaName = sourceAreaCountry;
							sourceAreaId = SourceArea.getCountryID()+"";
						}
						sourceAreaCityId = sourceAreaId;
						Sourcelon = String.valueOf(SourceArea.getLon());//经度
						Sourcelat = String.valueOf(SourceArea.getLat());//纬度

						if(!"".equals(SourceArea.getProviceID()) && SourceArea.getProviceID() !=0 ) {
							sourceAreaId = SourceArea.getProviceID()+"";
							areaName_s = SourceArea.getProviceName().replace("省", "");
							areaName_s = areaName_s.replace("市", "");
						} else {
							areaName_s = SourceArea.getCountryName();
							sourceAreaId = SourceArea.getCountryID()+"";
						}
						sourceAreaProvinceId = sourceAreaId;
					}   

					if(DestArea != null ) {    
						destAreaCountry = DestArea.getCountryName();
						destAreaCity = DestArea.getCityName();
						if(destAreaCity != null) {
							destAreaName = destAreaCity;
						} else {   
							destAreaName = destAreaCountry;
						}
						Destlon = String.valueOf(DestArea.getLon());
						Destlat = String.valueOf(DestArea.getLat());
					}  
					
//					Logger.getLogger(getClass()).info("SourceDataFilter ## " + dip+"#"+ sip +destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#" +areaName_s + "#" + sourceAreaCityId + "#" + sourceAreaProvinceId + "#" + sourceAreaCountry );
					if((sipLong >= 167772160 && sipLong <= 184549375) 
						|| (sipLong >= 2886729728L && sipLong <= 2887778303L)
						|| (sipLong >= 3232235520L &&sipLong <= 3232301055L)) {
						
//						if(DestArea != null) {
//							destAreaCountry = DestArea.getCountryName();
//							destAreaCity = DestArea.getCityName();
//							//						sourceAreaProvince = SourceArea.getProviceName();
//							if(destAreaCity != null) {
//								destAreaName  = destAreaCity;
//								destAreaId = DestArea.getCityID()+"";
//							} else {
//								destAreaName = destAreaCountry;
//								destAreaId = DestArea.getCountryID()+"";
//							}
//							destAreaCityId = destAreaId;
//							Destlon = String.valueOf(DestArea.getLon());
//							Destlat = String.valueOf(DestArea.getLat());
//
//							if(!"".equals(DestArea.getProviceID()) && DestArea.getProviceID() !=0 ) {
//								destAreaId = DestArea.getProviceID()+"";
//								areaName_d = DestArea.getProviceName().replace("省", "");
//								areaName_d = areaName_d.replace("市", "");
//							} else {
//								areaName_d = DestArea.getCountryName();
//								destAreaId = DestArea.getCountryID()+"";
//							}
//							destAreaProvinceId = destAreaId;
//						}   
//
//						if(SourceArea != null) {    
//							sourceAreaCountry = SourceArea.getCountryName();
//							sourceAreaCity = SourceArea.getCityName();
//							if(sourceAreaCity != null) {
//								sourceAreaName = sourceAreaCity;
//							} else {   
//								sourceAreaName = sourceAreaCountry;
//							}
//							Sourcelon = String.valueOf(SourceArea.getLon());//经度
//							Sourcelat = String.valueOf(SourceArea.getLat());//纬度
//						}  

						if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null) {
							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#" +areaName_s + "#" + sourceAreaCityId + "#" + sourceAreaProvinceId + "#" + sourceAreaCountry;
							gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
							ret.setFields("MetisSipStream", dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" +domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" +upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" +dnStreamOctets+ "|" +dnStreamPacket+ "|" + refer + "|" +post +"|OUT|" + appType +"|" +appId + "|" + accesstimeL);
							//							collector.emit("SipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno,upStreamOctets,upStreamPacket,status,userAgent,cookie,post,refer));
							//							collector.ack(tuple);
						}
					}else if((dipLong >= 167772160 && dipLong <= 184549375) 
						|| (dipLong >= 2886729728L && dipLong <= 2887778303L)
						|| (dipLong >= 3232235520L &&dipLong <= 3232301055L)) {
//						if(SourceArea != null) {
//							sourceAreaCountry = SourceArea.getCountryName();
//							sourceAreaCity = SourceArea.getCityName();
//							//						sourceAreaProvince = SourceArea.getProviceName();
//							if(sourceAreaCity != null) {
//								sourceAreaName  = sourceAreaCity;
//								sourceAreaId = SourceArea.getCityID()+"";
//							} else {
//								sourceAreaName = sourceAreaCountry;
//								sourceAreaId = SourceArea.getCountryID()+"";
//							}
//							sourceAreaCityId = sourceAreaId;
//							Sourcelon = String.valueOf(SourceArea.getLon());//经度
//							Sourcelat = String.valueOf(SourceArea.getLat());//纬度
//
//							if(!"".equals(SourceArea.getProviceID()) && SourceArea.getProviceID() !=0 ) {
//								sourceAreaId = SourceArea.getProviceID()+"";
//								areaName_s = SourceArea.getProviceName().replace("省", "");
//								areaName_s = areaName_s.replace("市", "");
//							} else {
//								areaName_s = SourceArea.getCountryName();
//								sourceAreaId = SourceArea.getCountryID()+"";
//							}
//							sourceAreaProvinceId = sourceAreaId;
//						}else if(DestArea != null ) {    
//							destAreaCountry = DestArea.getCountryName();
//							destAreaCity = DestArea.getCityName();
//							if(destAreaCity != null) {
//								destAreaName = destAreaCity;
//							} else {   
//								destAreaName = destAreaCountry;
//							}
//							Destlon = String.valueOf(DestArea.getLon());
//							Destlat = String.valueOf(DestArea.getLat());
//						}  

						if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null && areaName_s != null) {
							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#" +areaName_s + "#" + sourceAreaCityId + "#" + sourceAreaProvinceId + "#" + sourceAreaCountry;
							gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
							ret.setFields("MetisDipStream", dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" +domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" +upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" +dnStreamOctets+ "|" +dnStreamPacket+ "|" + refer + "|" +post+"|IN|" + appType +"|" +appId + "|" + accesstimeL);
							//							collector.emit("SipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno,upStreamOctets,upStreamPacket,status,userAgent,cookie,post,refer));
							//							collector.ack(tuple);
						}
					}
					if((IPSet == null || IPSet.contains(dip))) {
//						if(SourceArea != null) {
//							sourceAreaCountry = SourceArea.getCountryName();
//							sourceAreaCity = SourceArea.getCityName();
//							//						sourceAreaProvince = SourceArea.getProviceName();
//							if(sourceAreaCity != null) {
//								sourceAreaName  = sourceAreaCity;
//								sourceAreaId = SourceArea.getCityID()+"";
//							} else {
//								sourceAreaName = sourceAreaCountry;
//								sourceAreaId = SourceArea.getCountryID()+"";
//							}
//							sourceAreaCityId = sourceAreaId;
//							Sourcelon = String.valueOf(SourceArea.getLon());//经度
//							Sourcelat = String.valueOf(SourceArea.getLat());//纬度
//
//							if(!"".equals(SourceArea.getProviceID()) && SourceArea.getProviceID() !=0 ) {
//								sourceAreaId = SourceArea.getProviceID()+"";
//								areaName_s = SourceArea.getProviceName().replace("省", "");
//								areaName_s = areaName_s.replace("市", "");
//							} else {
//								areaName_s = SourceArea.getCountryName();
//								sourceAreaId = SourceArea.getCountryID()+"";
//							}
//							sourceAreaProvinceId = sourceAreaId;
//						}   
//
//						if(DestArea != null ) {    
//							destAreaCountry = DestArea.getCountryName();
//							destAreaCity = DestArea.getCityName();
//							if(destAreaCity != null) {
//								destAreaName = destAreaCity;
//							} else {   
//								destAreaName = destAreaCountry;
//							}
//							Destlon = String.valueOf(DestArea.getLon());
//							Destlat = String.valueOf(DestArea.getLat());
//						}  

						if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null && areaName_s != null) {
							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#" +areaName_s + "#" + sourceAreaCityId + "#" + sourceAreaProvinceId + "#" + sourceAreaCountry;
							gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
							ret.setFields("DipStream", dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" +domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" +upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" +dnStreamOctets+ "|" +dnStreamPacket+ "|" + refer + "|" +post+"|IN|" + appType +"|" +appId + "|" + accesstimeL);
							//							collector.emit("SipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno,upStreamOctets,upStreamPacket,status,userAgent,cookie,post,refer));
							//							collector.ack(tuple);
						}


					}else if (IPSet == null || IPSet.contains(sip)) {
//						if(DestArea != null) {
//							destAreaCountry = DestArea.getCountryName();
//							destAreaCity = DestArea.getCityName();
//							//						sourceAreaProvince = SourceArea.getProviceName();
//							if(destAreaCity != null) {
//								destAreaName  = destAreaCity;
//								destAreaId = DestArea.getCityID()+"";
//							} else {
//								destAreaName = destAreaCountry;
//								destAreaId = DestArea.getCountryID()+"";
//							}
//							destAreaCityId = destAreaId;
//							Destlon = String.valueOf(DestArea.getLon());
//							Destlat = String.valueOf(DestArea.getLat());
//
//							if(!"".equals(DestArea.getProviceID()) && DestArea.getProviceID() !=0 ) {
//								destAreaId = DestArea.getProviceID()+"";
//								areaName_d = DestArea.getProviceName().replace("省", "");
//								areaName_d = areaName_d.replace("市", "");
//							} else {
//								areaName_d = DestArea.getCountryName();
//								destAreaId = DestArea.getCountryID()+"";
//							}
//							destAreaProvinceId = destAreaId;
//						}   
//
//						if(SourceArea != null) {    
//							sourceAreaCountry = SourceArea.getCountryName();
//							sourceAreaCity = SourceArea.getCityName();
//							if(sourceAreaCity != null) {
//								sourceAreaName = sourceAreaCity;
//							} else {   
//								sourceAreaName = sourceAreaCountry;
//							}
//							Sourcelon = String.valueOf(SourceArea.getLon());//经度
//							Sourcelat = String.valueOf(SourceArea.getLat());//纬度
//						}  

						if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null) {
//							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + destAreaCountry + "#" + destAreaId + "#" +areaName_d + "#" + destAreaCityId + "#" + destAreaProvinceId + "#" + sourceAreaCountry;
							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#" +areaName_s + "#" + sourceAreaCityId + "#" + sourceAreaProvinceId + "#" + sourceAreaCountry;
							gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
							ret.setFields("SipStream", dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64
									+ "|" +domainname+ "|" +url + "|" +type+ "|" +delaytime+ "|" +idcno
									+ "|" +upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|"
									+dnStreamOctets+ "|" +dnStreamPacket+ "|" + refer + "|" +post+"|OUT|" + appType +"|"
									+appId + "|" + accesstimeL);
							//							collector.emit("SipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno,upStreamOctets,upStreamPacket,status,userAgent,cookie,post,refer));
							//							collector.ack(tuple);
						}
					}
				}
			}
		}
		return ret;
	}

}
